package ee.ut.math.chess.hadoop;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.TreeMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsAction;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;

import ee.ut.math.chess.base.model.Move;
import ee.ut.math.chess.game.model.Game;
import ee.ut.math.chess.hadoop.dummy.DummyHadoopManager;

public class HadoopFsManager {
	private static Configuration conf;

	static {
		conf = new Configuration();
		conf.set("mapred.job.tracker", "hdfs://node0:9001/");
		conf.set("fs.default.name", "hdfs://node0:9000/");
	}

	public static void mkdirInHadoop(String pathString) throws IOException {
		FileSystem fs = FileSystem.get(conf);
		Path path = new Path(pathString);
		fs.mkdirs(path);

		FsPermission permission = new FsPermission(FsAction.ALL, FsAction.ALL,
				FsAction.ALL);
		fs.setPermission(path, permission);
	}

	public static void deleteDirInHadoop(String pathString) throws IOException {
		FileSystem fs = FileSystem.get(conf);
		Path path = new Path(pathString);
		fs.delete(path, true);
	}

	public static void writeText(String pathStr) throws IOException {
		FileSystem fs = FileSystem.get(conf);

		Path path = new Path(pathStr);
		SequenceFile.Writer writer = SequenceFile.createWriter(fs, conf, path,
				Text.class, Text.class, SequenceFile.CompressionType.NONE);

		writer.append(new Text("1"), new Text("1"));
		writer.close();
	}
	

	public static java.util.Map<Long, Move> readMoves(String pathStr) throws IOException {
		java.util.Map<Long,Move> result = new TreeMap<Long,Move>();
		
		FileSystem fs = FileSystem.get(conf);
		Path path = new Path(pathStr);

		if (!fs.isFile(path)) {
			FileStatus[] listFiles = fs.listStatus(path);
			for (int i = 0; i < listFiles.length; i++) {
				try {
					SequenceFile.Reader reader = new SequenceFile.Reader(fs,
							listFiles[i].getPath(), conf);
					Text key = new Text();
					Text value = new Text();
					
					while (reader.next(key, value)) {
						result.put(Long.valueOf(value.toString()), new Move(key.toString()));
					}
					reader.close();
				} catch (Exception e) {
				}
			}
		}
		return result;
	}
	   

    public static java.util.Map<Double, List<Move>> readMoves2(String pathStr) throws IOException {
        java.util.Map<Double,List<Move>> result = new TreeMap<Double,List<Move>>();
        
        FileSystem fs = FileSystem.get(conf);
        Path path = new Path(pathStr);

        if (!fs.isFile(path)) {
            FileStatus[] listFiles = fs.listStatus(path);
            for (int i = 0; i < listFiles.length; i++) {
                try {
                    SequenceFile.Reader reader = new SequenceFile.Reader(fs,
                            listFiles[i].getPath(), conf);
                    Text key = new Text();
                    DoubleWritable value = new DoubleWritable();
                    
                    while (reader.next(key, value)) {
                    	
                    	List<Move> list = result.get(value.get());
						if(list == null){
							list = new ArrayList<Move>();
							list.add(new Move(key.toString()));
							result.put(value.get(), list);
						}else{
							list.add(new Move(key.toString()));
						}
                    }
                    reader.close();
                } catch (Exception e) {
                }
            }
        }
        return result;
    }
    
    public static void writeGame(String pathString, Game game) throws IOException {
        FileSystem fs = FileSystem.get(conf);

        Path path = new Path(pathString);
        SequenceFile.Writer writer = SequenceFile.createWriter(fs, conf, path,
                Text.class, BytesWritable.class, SequenceFile.CompressionType.NONE);

        writer.append(new Text("1"), new BytesWritable(DummyHadoopManager.getBytesFromGame(game)));
        writer.close();
    }
}
